package com.qf.workqueue;

import com.qf.util.ConnectionUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;

/**
 * work queue模式的消费者
 */
public class Consumer {

    @Test
    public void consume1() throws Exception {
        //1. Conn
        Connection connection = ConnectionUtil.getConnection();

        //2. Chann
        Channel channel = connection.openChannel().get();

        // 流控设置
        channel.basicQos(1);

        //3. callback
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1号，消费到队列消息：" + new String(body));
                System.out.println(envelope.getDeliveryTag());
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };

        //4. consumer
        channel.basicConsume("workqueue_queue",false,callback);

        //5. System.in
        System.in.read();

        //6. close
        channel.close();
        connection.close();
    }


    @Test
    public void consume2() throws Exception {
        //1. Conn
        Connection connection = ConnectionUtil.getConnection();

        //2. Chann
        Channel channel = connection.openChannel().get();

        channel.basicQos(5);

        //3. callback
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者2号，消费到队列消息：" + new String(body));
                System.out.println(envelope.getDeliveryTag());
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };

        //4. consumer
        channel.basicConsume("workqueue_queue",false,callback);

        //5. System.in
        System.in.read();

        //6. close
        channel.close();
        connection.close();
    }

}
